﻿using System.Diagnostics.Contracts;
using System.Reactive.Concurrency;
using System.Reactive.Subjects;

namespace System.Reactive.Linq
{
	public static partial class Observable2
	{
		internal static Func<T1, T2, T3, IObservable<Unit>> FromAsyncPattern<T1, T2, T3>(
			FuncExtended<T1, T2, T3, AsyncCallback, object, IAsyncResult> begin,
			Action<IAsyncResult> end)
		{
			Contract.Requires(begin != null);
			Contract.Requires(end != null);
			Contract.Ensures(Contract.Result<Func<T1, T2, T3, IObservable<Unit>>>() != null);

			return FromAsyncPattern<T1, T2, T3, Unit>(
				begin,
				result =>
				{
					end(result);

					return Unit.Default;
				});
		}

		internal static Func<T1, T2, T3, IObservable<TResult>> FromAsyncPattern<T1, T2, T3, TResult>(
			FuncExtended<T1, T2, T3, AsyncCallback, object, IAsyncResult> begin,
			Func<IAsyncResult, TResult> end)
		{
			Contract.Requires(begin != null);
			Contract.Requires(end != null);
			Contract.Ensures(Contract.Result<Func<T1, T2, T3, IObservable<TResult>>>() != null);

			return (arg1, arg2, arg3) =>
				{
					var subject = new AsyncSubject<TResult>();

					try
					{
						begin(
							arg1,
							arg2,
							arg3,
							result =>
							{
								TResult local;

								try
								{
									local = end(result);
								}
								catch (Exception exception)
								{
									subject.OnError(exception);

									return;
								}

								subject.OnNext(local);
								subject.OnCompleted();
							},
							null);
					}
					catch (Exception exception)
					{
						return Observable.Throw<TResult>(exception, Scheduler.ThreadPool);
					}

					return subject.AsObservable();
				};
		}
	}
}